package cn.hyxy.hadoop;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Demo10_MyInputFormat extends Configured implements Tool{

	@Override
	public int run(String[] args) throws Exception {
		if(args.length!=2) {
			System.out.println("usage ...");
			return -1;
		}
		// 7:声明Job
		Configuration config = getConf();
		FileSystem fs = FileSystem.get(config);
		Path path = new Path(args[1]);
		if(fs.exists(path)) {
			fs.delete(path, true);
		}
		// 8:Job
		Job job = Job.getInstance(config, "自定义输入类");
		
		job.setJarByClass(getClass());
		
		job.setMapperClass(MyMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(NullWritable.class);
		// 10:设置Reduce
		job.setNumReduceTasks(0);
		/*job.setReducerClass(MyReducer.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(LongWritable.class);*/
		// 设置读取的文件!!!!!!
		job.setInputFormatClass(MyLineInputFormat.class);
		MyLineInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, path);
		// 开始任务
		boolean boo = job.waitForCompletion(true);
		return boo ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {
		int code = ToolRunner.run(new Demo09_CNWordCountMR(), args);
		System.exit(code);
	}
	
	public static class MyMapper extends Mapper<Text, Text, Text, NullWritable>{
		@Override
		protected void map(Text key, Text value, Mapper<Text, Text, Text, NullWritable>.Context context)
				throws IOException, InterruptedException {
			key.set(key.toString()+"-----"+value.toString());
			context.write(key, NullWritable.get());
		}
	}
		
	public static class MyLineInputFormat extends FileInputFormat<Text, Text>{
		@Override
		public RecordReader<Text, Text> createRecordReader(InputSplit arg0, TaskAttemptContext arg1)
				throws IOException, InterruptedException {
			MyLineReader mReader=new MyLineReader();			
			return mReader;
		}
	}
		
	//2:开发一个读取文件的对象
	public static class MyLineReader extends RecordReader<Text, Text>{
		private BufferedReader br;
		private int line=1;
		private Text key=new Text();
		private Text value=new Text();		
		private float sum;
		
		@Override
		public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
			FileSplit fileSplit=(FileSplit) split;
			Path path=fileSplit.getPath();
//			br=new BufferedReader(new InputStreamReader(new FileInputStream(name))); //读取Windows文件的
			
			FileSystem fs=FileSystem.get(context.getConfiguration());
			InputStream in=fs.open(path);
			br=new BufferedReader(new InputStreamReader(in));
		}

		@Override
		public boolean nextKeyValue() throws IOException, InterruptedException {
			String line=br.readLine();
			if (line==null) {
				return false;
			}
			//设置key和value
			key.set("第["+(this.line++)+"行]");
			value.set(line);
			return true;
		}
		
		@Override
		public void close() throws IOException {
			if (br!=null) {
				br.close();
			}
		}

		@Override
		public Text getCurrentKey() throws IOException, InterruptedException {
			return key;
		}

		@Override
		public Text getCurrentValue() throws IOException, InterruptedException {
			return value;
		}

		@Override
		public float getProgress() throws IOException, InterruptedException {
			
			
			return 0;
		}
		
	}
	
	public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
		
		@Override
		protected void reduce(Text key3, Iterable<LongWritable> value3,
				Reducer<Text, LongWritable, Text, LongWritable>.Context context) throws IOException, InterruptedException {
			long sum = 0;
			for (LongWritable ii : value3) {
				sum = sum + ii.get();
			}
			context.write(key3, new LongWritable(sum));
		}
	}

}
